package com.witsight.cassandra.dao.impl;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;

import org.springframework.cassandra.core.CachedPreparedStatementCreator;
import org.springframework.cassandra.core.PreparedStatementBinder;
import org.springframework.cassandra.core.PreparedStatementCallback;
import org.springframework.cassandra.core.PreparedStatementCreator;
import org.springframework.cassandra.core.ResultSetExtractor;
import org.springframework.cassandra.core.RowCallback;
import org.springframework.cassandra.core.RowCallbackHandler;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.InvalidDataAccessApiUsageException;
import org.springframework.data.cassandra.convert.CassandraConverter;
import org.springframework.data.cassandra.core.CassandraTemplate;
import org.springframework.data.domain.Pageable;
import org.springframework.data.web.querydsl.QuerydslPredicateArgumentResolver;
import org.springframework.util.Assert;

import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.ColumnDefinitions;
import com.datastax.driver.core.PagingState;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.ColumnDefinitions.Definition;
import com.datastax.driver.core.exceptions.DriverException;
import com.witsight.cassandra.dao.CustomBoundStatement;
import com.witsight.cassandra.dao.PageCassandraOperations;
import com.witsight.platform.model.BoEntity;
import com.witsight.platform.util.lang.MapUtil;

/**   
 * 说明：自定义翻页CassandraTemplate
 * @Title: PageCassandraTemplate.java 
 * @Package com.witsight.cassandra.dao 
 * @See: {@link CassandraTemplate} {@link PageCassandraOperations}
 * Copyright: Copyright (c) 2017
 * Company:sany huax witsight team by product
 * @author: penght  
 * @date: 2017年9月27日 下午5:59:32 
 * @version: V1.0
 *
 */
public class PageCassandraTemplate extends CassandraTemplate implements PageCassandraOperations {
	/**
	 * Creates an instance of {@link PageCassandraTemplate} initialized with the given {@link Session}
	 * and {@link CassandraConverter}.
	 *
	 * @param session {@link Session} used to interact with Cassandra; must not be {@literal null}.
	 * @param converter {@link CassandraConverter} used to convert between Java and Cassandra types;
	 * must not be {@literal null}.
	 * @see org.springframework.data.cassandra.convert.CassandraConverter
	 * @see com.datastax.driver.core.Session
	 */
	public PageCassandraTemplate(Session session, CassandraConverter converter) {
		super(session, converter);
	}
	/**
	 * 
	 * 说明：根据检索条件检索数据
	 * @Title: query
	 * @See: {@link #query(PreparedStatementCreator, PreparedStatementBinder, RowCallbackHandler)}
	 * @author penght
	 * @param query 检索cql
	 * @param params 参数
	 * @param rowCallbackHandler 行回调
	 * @date: 2017年9月28日 下午3:47:35
	 */
	public void query(String query,final BoEntity params, RowCallbackHandler rowCallbackHandler) {
		Assert.hasText(query, "CQL must not be empty");
		
		super.query(new CachedPreparedStatementCreator(query), new PreparedStatementBinder() {

			@Override
			public BoundStatement bindValues(PreparedStatement prepared) throws DriverException {
				CustomBoundStatement bound = new CustomBoundStatement(prepared);
				if (MapUtil.isEmpty(params)) {
					return bound;
				}
				for (String key : params.keySet()) {
					if (!isCqlParam(bound, key)) {
						continue;
					}
					Object object = params.get(key);
					bound.set(key, object);
				}
				return bound;
			}
			
		}, rowCallbackHandler);
	}
	
	/**
	 * 
	 * 说明：根据检索条件检索数据 (ResultSet 接口返回)
	 * @Title: query
	 * @See: {@link #query(PreparedStatementCreator, PreparedStatementBinder, ResultSetExtractor)}
	 * @author penght
	 * @param query 检索cql
	 * @param params 参数
	 * @param resultSetExtractor 行回调
	 * @date: 2017年9月28日 下午3:47:35
	 */
	public <T> T query(String query,final BoEntity params, ResultSetExtractor<T> resultSetExtractor) {
		Assert.hasText(query, "CQL must not be empty");
		return super.query(new CachedPreparedStatementCreator(query), new PreparedStatementBinder() {

			@Override
			public BoundStatement bindValues(PreparedStatement prepared) throws DriverException {
				CustomBoundStatement bound = new CustomBoundStatement(prepared);
				if (MapUtil.isEmpty(params)) {
					return bound;
				}
				for (String key : params.keySet()) {
					if (!isCqlParam(bound, key)) {
						continue;
					}
					Object object = params.get(key);
					bound.set(key, object);
				}
				return bound;
			}
			
		}, resultSetExtractor);
	}
	/**
	 * 
	 * 说明：转换BoEntity
	 * @Title: toBoEntity
	 * @See: {@link #toBoEntity(Row)}
	 * @author penght
	 * @param row 行数据
	 * @return BoEntity
	 * @date: 2017年9月29日 上午7:57:19
	 */
	public BoEntity toBoEntity(Row row) {
		Assert.notNull(row, "Row must not be null");
		BoEntity map = null;

		if (row != null) {
			ColumnDefinitions columns = row.getColumnDefinitions();
			map = new BoEntity(columns.size());

			for (Definition columnDefinition : columns.asList()) {
				map.put(columnDefinition.getName(), columnToObject(row, columnDefinition));
			}
		}

		return map;
	}
	
	@SuppressWarnings("unchecked")
	private <T> T columnToObject(final Row row, final Definition columnDefinition) {
		return ((T) row.getObject(columnDefinition.getName()));
	}
	
//	
//	
//	/**
//	 * 
//	 * 说明：[]
//	 * @Title: select
//	 * @See: []
//	 * @author penght
//	 * @param statement
//	 * @param entityClass
//	 * @return 
//	 * @date: 2017年9月28日 下午2:31:54
//	 */
//	public <T> List<T> select(final Statement statement, Class<T> entityClass) {
//		Assert.notNull(statement, "CQL must not be empty");
//		Assert.notNull(entityClass, "EntityClass must not be null");
//		return processResultSet(executeStatement(statement), new CassandraConverterRowCallback<T>(cassandraConverter, entityClass));
//	}
//	
//	public <T> List<T> processResultSet(ResultSet resultSet, RowCallback<T> rowCallback) {
//		List<T> result = new ArrayList<T>();
//
//		for (Row row : EmptyResultSet.nullSafeResultSet(resultSet)) {
//			result.add(rowCallback.doWith(row));
//		}
//
//		return result;
//	}
//	
//	/**
//	 * Execute a command at the Session Level with optional options
//	 *
//	 * @param statement The query to execute.
//	 */
//	public ResultSet executeStatement(final Statement statement) {
//		return super.doExecute(statement);
//	}
//	
	private boolean isCqlParam(BoundStatement bound, String key) {
		boolean flg = true;
		try {
			bound.isNull(key);
		} catch (Exception e) {
			flg = false;
		}
		
		return flg;
	}
//	/**
//	 * 
//	 * 说明：动态的创建参数
//	 * @Title: createStatement
//	 * @See: {@link #createStatement(String, BoEntity)}
//	 * @author penght
//	 * @param cql cql query语句
//	 * @param params 参数
//	 * @return Statement
//	 * @date: 2017年9月28日 下午2:15:19
//	 */
//	public Statement createStatement(String cql, BoEntity params) {
//		Assert.hasText(cql, "cql must be not empty.");
//		Assert.notEmpty(params, "values must be not empty.");
//		PreparedStatement prepared = super.getSession().prepare(cql);
//		CustomBoundStatement bound = new CustomBoundStatement(prepared);
//		for (String key : params.keySet()) {
//			if (!this.isCqlParam(bound, key)) {
//				continue;
//			}
//			Object object = params.get(key);
//			bound.set(key, object);
//		}
//		
//		return bound;
//	}
	
//	public ResultSet skipRows(String cql, BoEntity params, int start, int size) {
//		
//	}
	
	private BoundStatement getBoundStatement(PreparedStatement prepared,final BoEntity params) {
		CustomBoundStatement bound = new CustomBoundStatement(prepared);
		if (MapUtil.isEmpty(params)) {
			return bound;
		}
		for (String key : params.keySet()) {
			if (!isCqlParam(bound, key)) {
				continue;
			}
			Object object = params.get(key);
			bound.set(key, object);
		}
		return bound;
	}
	/**
	 * 
	 * 说明：检索总数
	 * @Title: selectCount
	 * @See: {@link #selectCount(String)}
	 * @author penght
	 * @param countCql 
	 * @return 
	 * @date: 2017年9月29日 上午8:34:26
	 */
	public long selectCount(String countCql) {
		Assert.hasText(countCql, "CQL must not be empty");
		return super.query(countCql,  new ResultSetExtractor<Long>() {

			@Override
			public Long extractData(ResultSet resultSet) {

				Row row = resultSet.one();

				if (row == null) {
					throw new InvalidDataAccessApiUsageException(
							String.format("count query [%1$s] did not return any results", countCql));
				}

				return row.getLong(0);
			}
		});
	}
	/**
	 * 
	 * 说明：根据检索条件检索数据page(ResultSet 接口返回)
	 * @Title: query
	 * @See: {@link #query(PreparedStatementCreator, PreparedStatementBinder, ResultSetExtractor)}
	 * @author penght
	 * @param query 检索cql
	 * @param params 参数
	 * @param resultSetExtractor 行回调
	 * @date: 2017年9月28日 下午3:47:35
	 */
	public <T> T queryPage(String query,final BoEntity params, Pageable pageable, ResultSetExtractor<T> resultSetExtractor) {
		Assert.hasText(query, "CQL must not be empty");
		Assert.notNull(pageable, "Pageable must not be null");
		Assert.notNull(resultSetExtractor, "ResultSetExtractor must not be null");
		int pageSize = pageable.getPageSize();
		int start = (pageable.getPageNumber() * pageable.getPageSize()) + 1;
		return super.execute(new CachedPreparedStatementCreator(query), new PreparedStatementCallback<T>() {

			@Override
			public T doInPreparedStatement(PreparedStatement ps) throws DriverException, DataAccessException {
				BoundStatement bound = getBoundStatement(ps, params);
				return resultSetExtractor.extractData(skipRows(bound, start, pageSize));
			}
			
		});
	}
	
    private List<Row> getRows(ResultSet result, int start, int size) {
        List<Row> rows = new ArrayList<>(size);
        if (null == result) {
            return rows;
        }
        int skippingRows = (start - 1) % size;
        int index = 0;
        for (Iterator<Row> iter = result.iterator(); iter.hasNext() && rows.size() < size;) {
            Row row = iter.next();
            if (index >= skippingRows) {
                rows.add(row);
            }
            index ++;
        }
        return rows;
    }
	/**
	 * 
	 * 说明：翻页检索
	 * @Title: queryPage
	 * @See: {@link QuerydslPredicateArgumentResolver}
	 * @author penght
	 * @param query  检索条件
	 * @param params 参数
	 * @param pageable 翻页对象
	 * @param rowCallback 行回调
	 * @return 
	 * @date: 2017年10月9日 下午2:39:24
	 */
	public <T> List<T> queryPage(String query,final BoEntity params, Pageable pageable, RowCallback<T> rowCallback) {
		Assert.hasText(query, "CQL must not be empty");
		Assert.notNull(pageable, "Pageable must not be null");
		Assert.notNull(rowCallback, "RowCallback must not be null");
		int pageSize = pageable.getPageSize();
		int start = (pageable.getPageNumber() * pageable.getPageSize()) + 1;
		List<T> page = super.execute(new CachedPreparedStatementCreator(query), new PreparedStatementCallback<List<T>>() {

			@Override
			public List<T> doInPreparedStatement(PreparedStatement ps) throws DriverException, DataAccessException {
				BoundStatement bound = getBoundStatement(ps, params);
				ResultSet resultSet = skipRows(bound, start, pageSize);
				if (null == resultSet) {
					return Collections.<T> emptyList();
		        }
				List<T> rows = new ArrayList<>(pageSize);
				int skippingRows = (start - 1) % pageSize;
		        int index = 0;
		        for (Iterator<Row> iter = resultSet.iterator(); iter.hasNext() && rows.size() < pageSize;) {
		            Row row = iter.next();
		            if (index >= skippingRows) {
		            		rows.add(rowCallback.doWith(row));
		            }
		            index ++;
		        }
		        return rows;
			}
			
		});
		
		return page;
	}
	
	
	/**
	 * 
	 * 说明：翻页跳行方法
	 * @Title: skipRows
	 * @See: {@link #skipRows(Statement, int, int)}
	 * @author penght
	 * @param statement 检索statement
	 * @param start 开始
	 * @param size 结束
	 * @return 结果集合
	 * @date: 2017年9月28日 上午8:42:43
	 */
	public ResultSet skipRows(Statement statement, int start, int size) {
        ResultSet result = null;
        int skippingPages = getPageNumber(start, size);
        String savingPageState = null;
        statement.setFetchSize(size);
        boolean isEnd = false;
        for (int i = 0; i < skippingPages; i++) {
            if (null != savingPageState) {
                statement = statement.setPagingState(PagingState.fromString(savingPageState));
            }
            result = super.doExecute(statement);
            PagingState pagingState = result.getExecutionInfo().getPagingState();
            if (null != pagingState) {
                savingPageState = result.getExecutionInfo().getPagingState().toString();
            }

            if (result.isFullyFetched() && null == pagingState) {

                //if hit the end more than once, then nothing to return, otherwise, mark the isEnd to 'true'
                if (true == isEnd) {
                    return null;
                } else {
                    isEnd = true;
                }
            }
        }
        return result;
    }

    private int getPageNumber(int start, int size) {
        if (start < 1) {
            throw new IllegalArgumentException("Starting row need to be larger than 1");
        }
        int page = 1;
        if (start > size) {
            page = (start - 1) / size + 1;
        }
        return page;
    }
}
